package com.demo.config;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class KafKaConsumer {

	/**
	 	可以同时订阅多主题，只需按数组格式即可，也就是用“，”隔开
		如果多个集群的情况下，需要在KafkaListener监听注解上添加containerFactory，对应配置中的监听容器工厂。
	 */
//	@KafkaListener(topics = { "foo" }, containerFactory = "kafkaListenerContainerFactory")
	@KafkaListener(topicPartitions = @TopicPartition(
			topic = "foo",
			partitionOffsets = @PartitionOffset(
					partition = "0" ,
					initialOffset = "0")), containerFactory = "kafkaListenerContainerFactory")
	public void receive(ConsumerRecord<?, ?> record) {
		System.out.println("foo消费得到的消息---key: " + record.key() + " , " + record.value().toString());
	}

	@KafkaListener(topicPartitions = @TopicPartition(
			topic = "bar",
			partitionOffsets = @PartitionOffset(
					partition = "0" ,
					initialOffset = "0")), containerFactory = "kafkaListenerContainerFactory")
	public void barReceive(ConsumerRecord<?, ?> record) {
		System.out.println("bar消费得到的消息---key: " + record.key() + " , " + record.value().toString());
	}

//	    @KafkaListener(//id是消费者监听容器
//            topicPartitions =//配置topic和分区：监听两个topic，分别为topic1、topic2，topic1只接收分区0，3的消息，
//                    //topic2接收分区0和分区1的消息，但是分区1的消费者初始位置为5
//                    { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
//                      @TopicPartition(topic = "topic2", partitions = "0",
//                                    partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "0"))
//                    })
	@KafkaListener(//id是消费者监听容器
            topicPartitions =//配置topic和分区：监听两个topic，分别为topic1、topic2，topic1只接收分区0，3的消息，
                    //topic2接收分区0和分区1的消息，但是分区1的消费者初始位置为5
                    { @TopicPartition(topic = "topic3", partitionOffsets = @PartitionOffset(
							partition = "0" ,
							initialOffset = "0")),
                      @TopicPartition(topic = "topic4", partitionOffsets = @PartitionOffset(
							  partition = "0" ,
							  initialOffset = "0"))
                    })
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            System.out.println("topic" + record.topic());
            System.out.println("key:" + record.key());
            System.out.println("value:" + record.value());
        }
    }

	//    @KafkaListener(id = "myContainer2",topics = {"foo","bar"})
//	@KafkaListener(id = "foo", topicPartitions = @TopicPartition(
//			topic = "foo",
//			partitionOffsets = @PartitionOffset(
//					partition = "0" ,
//					initialOffset = "0")))
//	public void listen2(ConsumerRecord<?, ?> record){
//		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//		if (kafkaMessage.isPresent()) {
//			System.out.println("topic：" + record.topic());
//			System.out.println("key:" + record.key());
//			System.out.println("value:" + record.value());
//		}
//	}

}